Context

TestSSHHook::test_command_timeout_fail was intermittently failing in CI.

The original test opened a real SSH connection to ssh_default, ran sleep 1 with a 1 ms (0.001) timeout, and expected AirflowException to be raised. Two independent timeout mechanisms were racing:

1. paramiko's channel-level socket timeout.
   exec_command(timeout=0.001) calls Channel.settimeout(0.001), which makes every subsequent recv / send on the channel raise socket.timeout after 1 ms. This fires inside paramiko's own transport thread and during any blocking read the channel performs internally.

2. Airflow's select-loop timeout.
   exec_ssh_client_command passes cmd_timeout as the fourth argument to select([channel], [], [], cmd_timeout). When select returns an empty list, the function sets timedout = True and eventually raises AirflowException("SSH command timed out").

With a 1 ms deadline both mechanisms fire at roughly the same instant. Depending on thread scheduling, CPU load, and kernel select granularity:

• Happy path (test passes): select returns first with an empty read list, Airflow's own timeout logic triggers, and the test catches AirflowException.
• Unhappy path (test fails): paramiko's channel-level socket.timeout fires first—inside the transport thread or during the stdin.close() / channel.shutdown_write() calls that happen before the select loop is even reached. That surfaces as paramiko.ssh_exception.SSHException or socket.timeout, neither of which matches the AirflowException the test expects.

Because the race depends on real wall-clock time, it is non-deterministic. A fast CI runner tips the odds one way; a loaded one tips them the other.

What the fix does

The fix removes the real SSH connection entirely and replaces it with mocks that deterministically exercise the timeout detection logic inside exec_ssh_client_command.

The key mock is on select:

def fake_select(rlist, wlist, xlist, timeout=None):
    assert timeout == pytest.approx(0.001)
    return [], [], []

Returning ([], [], []) simulates select reporting "nothing readable within the timeout window." The production code then follows its normal path:

readq, _, _ = select([channel], [], [], cmd_timeout)
if cmd_timeout is not None:
    timedout = not readq          # True, because readq is []
...
if ... or timedout:
    stdout.channel.shutdown_read()
    stdout.channel.close()
    break
...
if timedout:
    raise AirflowException("SSH command timed out")

There is no second timeout mechanism in play. paramiko.SSHClient is a MagicMock(spec=...), so exec_command returns instantly with mock objects. stdin.close() and channel.shutdown_write() are no-ops on the mock, so they can never raise a socket-level exception. The only timeout that fires is Airflow's, which is exactly what the test is verifying.

What the fix verifies beyond the original test

The original test had a single assertion: "an AirflowException is raised." The new test adds:

All mocks use spec= against the real paramiko types (paramiko.Channel, paramiko.ChannelFile, paramiko.ChannelStdinFile, paramiko.ChannelStderrFile, paramiko.SSHClient), so accessing a misspelled attribute on any mock will raise AttributeError immediately rather than silently returning a new MagicMock.

What the fix does NOT do

It does not test the data-reading path (stdout/stderr aggregation), the graceful-exit path (command finishes before timeout), or the channel.close() race condition handler (lines 492-498). Those are separate behaviors covered by other tests (test_command_timeout_success, test_command_timeout_not_set, and the broader exec_ssh_client_command integration tests). This test is scoped to one thing: the timeout detection and cleanup path.

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Opus 4.6 following the guidelines

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.